package com.camemax.utils;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.Properties;

/*
* Flink读取硬盘上的文件记录，逐行读入Kafka
* */
public class ReadFileUtils {
    public static void readFile2Kafka(String filePath, Properties prop, String topic) throws IOException {
        // 1. Kafka配置信息
//        Properties props = new Properties();
//        props.setProperty("bootstrap.servers", "172.28.40.190:9092");
//        props.setProperty("group.id",groupId);


        // 2. Config Kafka Producer
        KafkaProducer<String, String> producer = new KafkaProducer<>(prop);

        // 3. Config buffer reading[BufferedReader]
        BufferedReader readFileBuffer = new BufferedReader(new FileReader(filePath));
        String dataLine;
        while( (dataLine = readFileBuffer.readLine()) != null ){
            // Config a ProducerRecord[record] for the line from the target file[dataLine]
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, dataLine);
            // Kafka Producer Send ProducerRecord
            producer.send(record);
        }
        readFileBuffer.close();
    }
}
